/**
 * 2017年6月29日
 */
package org.textplantform.common.core.mq;

import java.util.Arrays;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.apache.kafka.common.errors.WakeupException;
import org.textplantform.common.api.mq.MessageConsumer;
import org.textplantform.common.api.mq.PlantformTextConsumer;
import org.textplantform.common.core.LoggerSupport;

/**
 * @author Alex
 *
 */
public abstract class BasePlantformKafkaConsumer extends LoggerSupport implements PlantformTextConsumer<ConsumerRecord<String,String>>{
	protected MessageConsumer<ConsumerRecord<String,String>> mc;
	protected KafkaConsumer<String,String> kc;
	protected static ExecutorService defaultES = null;
	private volatile boolean running = true;
	protected String topic = null;
	protected String [] topics = null;
	public BasePlantformKafkaConsumer(KafkaConsumer<String,String> kc,String ... topics) {
		this.kc = kc;
		this.topics = topics;
	}
	private final void setSubsc(){
		kc.subscribe(Arrays.asList(topics));
	}
	private void initES(){
		if(defaultES == null){
			synchronized (BasePlantformKafkaConsumer.class) {
				if(defaultES == null){
					defaultES =  Executors.newCachedThreadPool();
				}
			}
		}
	}
	@Override
	public final void startAsync() {
		initES();
		defaultES.execute(()->{
			startSync();
		});
	}
	@Override
	public final void startAsync(ExecutorService es) {
		es.execute(()->{
			startSync();
		});
	}
	
	@Override
	public final void startSync() {
		setSubsc();
		running = true;
		while(running){
			try{
				 ConsumerRecords<String, String> records = kc.poll(50);
				 if(records.count() == 0) continue;
				 consume(records);
			}catch(WakeupException e){
				logger.warn("wake up exception catched ,now consumer exit ..");
				running = false;
				break;
			}catch(Exception e){
				logException(e);
			}
		}
	}
	public final void stop(){
		this.running = false;
		kc.wakeup();
		
	}
	public final void setMessageConsuemr(MessageConsumer<ConsumerRecord<String,String>> mc){
		this.mc = mc;
	}
	@Override
	public void consume(ConsumerRecord<String, String> message) {
		if(mc == null)
			System.out.println(message);
		else{
			mc.consume(message);
		}
	}
	@Override
	public void consume(Iterable<ConsumerRecord<String, String>> messages) {
		if(mc == null){
			for(ConsumerRecord<String, String> cr : messages){
				consume(cr);
			}
		}else{
			mc.consume(messages);
		}
		
	}

}
